[1172]python操作odps

#[1172]python操作odps| 来源: 网络整理| 查看: 265

文章目录 1、安装ODPS2、连接阿里云odps3、执行sql4、odps全表扫描在Python中怎么实现设置运行参数设置读取数据规模设置读取结果为pandas DataFrame设置读取速度（进程数）

PyODPS是MaxCompute的Python SDK，提供DataFrame框架和MaxCompute对象的基本操作方法。

接口手册：https://pyodps.readthedocs.io/zh_CN/latest/?spm=a2c4g.11186623.0.0.1aaf3d94n84mIN

PySdk下载：https://github.com/aliyun/aliyun-odps-python-sdk

1、安装ODPS pip install ODPS 2、连接阿里云odps

确认下载好后，连接odps，账号密码自备。

from odps import ODPS o = ODPS( access_id='user', #登陆账号 secret_access_key='password', #登陆密码 project='project', #odps上的项目名称 endpoint='http://service.cn-hangzhou-xxx:80/api') #官方提供的接口 3、执行sql

3.1、简单执行sql

with o.execute_sql('desc tablename').open_reader() as reader: print(reader.raw) o.execute_sql('select * from table_name') # 同步的方式执行，会阻塞直到SQL执行完成。 instance = o.run_sql('select * from table_name') # 异步的方式执行。 print(instance.get_logview_address()) # 获取logview地址 instance.wait_for_success() # 阻塞直到完成。

3.2、通过odps内置DataFrame读取，该方法读取的数据结构类型为odps.df.expr.core.DataFrame

def get_odps_table(tb_name): data = DataFrame(o.get_table(tb_name)) data['ds'] = data['ds'].astype('int') return data rdata = get_odps_table('tb_name') #获取表数据实例

3.3、封装成函数连接

直接输入sql就可进行增删改查

def exe_sql(sql): from datetime import datetime import pandas as pd st_time = datetime.now() data = [] with o.execute_sql(sql).open_reader() as reader: d = defaultdict(list) # collection默认一个dict # 解析record中的每一个元组,存储方式为(k,v),以k作为key,存储每一列的内容； for record in reader: for res in record: d[res[0]].append(res[1]) data = pd.DataFrame.from_dict(d, orient='index').T # 转换为数据框,并转置,不转置的话是横条数据 ed_time = datetime.now() print('总耗时：', ed_time - st_time ) # 输出sql运行时间，方便sql修改、维护 return data

3.4、将sql结果转换为excle

test = exe_sql('输入sql') test.to_excel('./表名', index=False, encoding='gbk') 4、odps全表扫描在Python中怎么实现

odps很重要的一个点是分区的概念，由于分区的存在会使得查询的时候数据量少很多，查询效率更高，但是有的时候分区的数据满足不了查询，需要全表查询的时候，在odpsSQL执行窗口中可以用这条语句来默认全表扫描，

SET odps.sql.allow.fullscan=true;

但是在Python中，如果这条语句直接放在SQL中，则执行的时候会报错，我们可以通过以下办法来解决

from odps import options options.sql.settings = {"odps.sql.submit.mode": "script"}

设置一下模式，这样就可以了

设置运行参数

在运行时如果需要设置参数，您可以通过设置hints参数来实现，参数的类型是dict。

o.execute_sql('select * from pyodps_iris', hints={'odps.sql.mapper.split.size': 16})

您可以对于全局配置设置sql.settings，后续每次运行时则都会添加相关的运行时参数。

from odps import options options.sql.settings = {'odps.sql.mapper.split.size': 16} o.execute_sql('select * from pyodps_iris') # 会根据全局配置添加hints 设置读取数据规模

如果您想要限制下载数据的规模，可以为open_reader增加limit选项，或者设置 options.tunnel.limit_instance_tunnel = True 。如果未设置 options.tunnel.limit_instance_tunnel，MaxCompute会自动打开数据量限制，此时，可下载的数据条数受到Project配置的Tunnel下载数据规模数限制，通常该限制为10000条。

设置读取结果为pandas DataFrame # 直接使用 reader 的 to_pandas 方法 with o.execute_sql('select * from dual').open_reader(tunnel=True) as reader: # pd_df 类型为 pandas DataFrame pd_df = reader.to_pandas() 设置读取速度（进程数）

您可以通过n_process指定使用进程数。

说明：目前多进程加速在 Windows 下无法使用。

import multiprocessing n_process = multiprocessing.cpu_count() with o.execute_sql('select * from dual').open_reader(tunnel=True) as reader: # n_process 指定成机器核数 pd_df = reader.to_pandas(n_process=n_process)

参考：https://help.aliyun.com/document_detail/34615.html https://blog.csdn.net/T_qiuqiu/article/details/126348157 https://www.jianshu.com/p/cc7d0461f6d5 https://blog.csdn.net/weixin_38938108/article/details/125853375 https://blog.csdn.net/weixin_42261073/article/details/103199562

实例操作：https://help.aliyun.com/document_detail/27830.html MaxCompute客户端常用命令：https://blog.csdn.net/yitian_z/article/details/105540527 pyodps：https://pyodps.readthedocs.io/en/latest/base-instances.html?highlight=instance 使用Logview 2.0查看Job运行信息：https://help.aliyun.com/document_detail/183576.html

【本文地址】

公司简介

联系我们